package com.doit.day02;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.sql.*;
import java.time.Duration;
import java.util.Arrays;
import java.util.Collection;
import java.util.Properties;

public class Kafka2Mysql {
    public static void main(String[] args) throws SQLException {
        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"linux01:9092");
        props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"G001");
        //可选
        props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
        props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false");

        KafkaConsumer<String, String> consume = new KafkaConsumer<String, String>(props);

        //jdbc对象
        Connection conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/test", "root", "123456");
        PreparedStatement pps = conn.prepareStatement("insert into user_info values(?,?,?,?)");
        PreparedStatement pps_offset = conn.prepareStatement("INSERT INTO consumer_offsets VALUES (?,?) ON DUPLICATE KEY UPDATE offset = ?");

        //再写一个sql来执行，去mysql数据库中查询数据(上次消费到哪了)
        PreparedStatement pps_get_offset = conn.prepareStatement("select offset from consumer_offsets where g_t_p = ?");
        //将原来自动提交事务改掉，改成手动提交
        conn.setAutoCommit(false);

        //读取mysql表中记录的偏移量，然后指定从什么地方开始往后继续读数据  ==》 做到，指定从什么偏移量开始消费数据

        /**
         * 现在考虑到，如果消费者组里面多了一个消费者
         * 那么，就需要发生消费者组在均衡
         * 我怎么知道我这个消费者到底是现在消费那几个分区的数据呢？
         */
        consume.subscribe(Arrays.asList("mysqlData"), new ConsumerRebalanceListener() {
            //回收前的所有分区
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {

            }

            //重新分配之后的所有分区
            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                for (TopicPartition topicPartition : partitions) {
                    try {
                        int partition = topicPartition.partition();
                        String topic = topicPartition.topic();
                        String g_t_p = "G001_"+topic+"_"+partition;
                        pps_get_offset.setString(1,g_t_p);
                        ResultSet resultSet = pps_get_offset.executeQuery();
                        if (resultSet.next()){
                            long offset = resultSet.getLong(1);
                            consume.seek(topicPartition,offset+1);
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        });

/*

        TopicPartition mysqlData1 = new TopicPartition("mysqlData", 0);
        TopicPartition mysqlData2 = new TopicPartition("mysqlData", 1);
        TopicPartition mysqlData3 = new TopicPartition("mysqlData", 2);

        consume.assign(Arrays.asList(mysqlData1,mysqlData2,mysqlData3));

        ResultSet resultSet = pps_get_offset.executeQuery();
        if (!resultSet.next()){
            consume.seek(mysqlData1,0);
            consume.seek(mysqlData2,0);
            consume.seek(mysqlData3,0);
        }else {
            do{
                String gtp = resultSet.getString(1);
                long offset = resultSet.getLong(2);
                if (gtp.endsWith("0")){
                    consume.seek(mysqlData1,offset+1);
                }else if(gtp.endsWith("1")){
                    consume.seek(mysqlData2,offset+1);
                }else {
                    consume.seek(mysqlData3,offset+1);
                }
            }while (resultSet.next());
        }
*/

        while (true){
            ConsumerRecords<String, String> poll = consume.poll(Duration.ofMillis(Long.MAX_VALUE));
            for (ConsumerRecord<String, String> record : poll) {
                try {
                    //业务逻辑的处理  1，zss，18，male  ==》 id name age  gender
                    String str = record.value();
                    //对字符串进行切割
                    String[] arr = str.split(",");
                    //写到数据库里面
                    pps.setInt(1,Integer.parseInt(arr[0]));
                    pps.setString(2,arr[1]);
                    pps.setInt(3,Integer.parseInt(arr[2]));
                    pps.setString(4,arr[3]);
                    //执行插入语句
                    pps.execute();

                    if(arr[0].equals("4")){
                        throw new Exception("老子自己造的异常，你不服来打我");
                    }


                    //手动提交偏移量
                    String topic = record.topic();
                    int partition = record.partition();
                    long offset = record.offset();

                    pps_offset.setString(1,"G001_"+topic+"_"+partition);
                    pps_offset.setLong(2,offset);
                    pps_offset.setLong(3,offset);
                    pps_offset.execute();
                    //提交事务
                    conn.commit();
                } catch (Exception e) {
                    System.out.println("打印一下错误，以示尊重："+e);
                    //回滚事务
                    conn.rollback();
                }
            }
        }



    }
}
